package com.yesun.task.pipeline.handler.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yesun.core.util.FileUtils;
import com.yesun.server.mdp.service.client.impl.DefaultClient;
import com.yesun.task.pipeline.common.Constant;
import com.yesun.task.pipeline.common.StockCode;
import com.yesun.task.pipeline.domain.dto.JobDto;
import com.yesun.task.pipeline.domain.dto.TaskDto;
import com.yesun.task.pipeline.handler.Executor;
import com.yesun.task.pipeline.handler.Handler;
import com.yesun.task.pipeline.handler.Loader;
import com.yesun.task.pipeline.worker.ParallelWorker;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

//分时分笔数据Handler
@Slf4j
public class DetailHandler extends Handler {

    static DefaultClient client = new DefaultClient();
    static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
    static SimpleDateFormat shortSdf = new SimpleDateFormat("yyyyMMdd");
    static String TODAY = shortSdf.format(Calendar.getInstance().getTime());
    static Loader loader;

    public DetailHandler(JobDto jobDto) {
        super(jobDto);
    }

    @Override
    public Loader getLoader() {
        if(loader == null) {
            //实例化
            loader = new Loader() {
                @Override
                public void load(BlockingQueue queue) {
                    log.info("加载分时分笔子任务...");
                    //实现资金流入子任务，插入到队列
                    String[] arr = StockCode.STOCK_CODE.split(",");
                    for(String code : arr) {
                        //update
                        long currentTime = System.currentTimeMillis();
                        String secid = "";
                        if(code.startsWith("6")) {
                            secid = "1." + code;
                        } else if(code.startsWith("0") || code.startsWith("3") || code.startsWith("8")) {
                            secid = "0." + code;
                        } else {
                            //暂时不支持其他股票
                            log.info("暂时不支持的股票类型：" + code);
                            continue;
                        }
                        String url = "https://86.push2.eastmoney.com/api/qt/stock/details/sse?fields1=f1%2Cf2%2Cf3%2Cf4&fields2=f51%2Cf52%2Cf53%2Cf54%2Cf55&mpi=2000&ut=bd1d9ddb04089700cf9c27f6f7426281&fltt=2&pos=-0&secid="+secid+"&wbp2u=%7C0%7C0%7C0%7Cweb";
                        TaskDto taskDto = new TaskDto();
                        taskDto.setJobName(jobDto.getJobName());
                        taskDto.setName(code + "分时分笔采集任务");
                        taskDto.setTag(Constant.JOB_TAG_DETAIL);

                        Map<String,String> map = new HashMap<>();
                        map.put("url", url);
                        map.put("code", code);
                        map.put("saveTo", "/df_detail/"+ shortSdf.format(new Date()) +"/"+code+".csv");//暂时只支持保存到磁盘
                        taskDto.setMap(map);


                        //加入MAP_TASK
                        ParallelWorker.MAP_TASK.put(taskDto.getName(), taskDto.getJobName());

                        queue.offer(taskDto);
                        log.info("添加子任务：" + taskDto.getName());
                    }
                }
            };
        }
        return loader;
    }

    @Override
    public Executor getExecutor() {
        return new Executor() {
            @Override
            public void execute(TaskDto taskDto) {
                //执行task的采集
                log.info("execute：" + taskDto.getName());
                long start = System.currentTimeMillis();
                try {
                    StringBuffer stringBuffer = new StringBuffer();
                    //由于数据源是一个event-stream接口，所以单独实现
                    CloseableHttpClient httpclient = HttpClients.createDefault();
                    CloseableHttpResponse response = null;
                    HttpGet httpGet = new HttpGet(taskDto.getMap().get("url"));
                    try {
                        response = httpclient.execute(httpGet);
                        HttpEntity entity = response.getEntity();
                        Header[] headers = response.getAllHeaders();
                        if (entity != null) {
                            InputStream inputStream = entity.getContent();
                            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
                            String line;
                            if ((line = reader.readLine()) != null) {
                                // 处理 event stream 数据
                                stringBuffer.append(line);
                            }
                        }
                    } catch (org.apache.http.client.ClientProtocolException e) {
                        e.printStackTrace();
                    } catch (IOException e) {
                        e.printStackTrace();
                    } finally {
                        if(response!=null) {
                            try {
                                response.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }

                    long requestEnd = System.currentTimeMillis();
                    log.info("http request time:" + (requestEnd-start));
                    String content = stringBuffer.toString();

                    content = content.substring(content.indexOf("{"), content.lastIndexOf("}")+1);
                    log.debug("content:"+content);
                    if (StringUtils.isEmpty(content)) {
                        log.info("content is empty:" + taskDto.getMap().get("url"));
                        return;
                    }

                    /** 数据结构
                     data: {"rc":0,"rt":12,"svr":2887007235,"lt":1,"full":1,"dlmkts":"",
                     "data":{"code":"300212","market":0,"decimal":2,"prePrice":31.73,
                     "details":
                     ["09:15:03,31.83,5,0,4",  #后面0表示笔数，和4成交类型：1卖出、2 买入、
                     "09:15:12,31.73,19,0,4",
                     "09:15:21,31.73,19,0,4",
                     "09:15:48,31.73,24,0,4",
                     **/

                    StringBuffer sb = new StringBuffer("采集日期,股票代码,时间,价格,交易量,笔数,交易类型\n"); //股价最好从其他表获取
                    JSONObject objRoot = JSONObject.parseObject(content);
                    if(objRoot != null) {
                        JSONObject objData = objRoot.getJSONObject("data");
                        if(objData != null) {
                            JSONArray jsonArray = objData.getJSONArray("details");
                            if (jsonArray != null) {
                                jsonArray.stream()
                                        .forEach(line -> {
                                            String[] arr = line.toString().split(",");
                                            sb.append(TODAY + "," + taskDto.getMap().get("code") + "," + arr[0] + "," + arr[1] + "," +  arr[2] + "," + arr[3] + "," + arr[4] +",\n");
                                        });
                                long processEnd = System.currentTimeMillis();
                                log.info("process content time:" + (processEnd-requestEnd));

                                //saveto
                                FileUtils.saveCsv(Constant.BASE_PATH + taskDto.getMap().get("saveTo"), sb.toString(), "UTF-8");
                                long saveEnd = System.currentTimeMillis();
                                log.info("save csv file time:" + (saveEnd-processEnd));

                            } else {
                                log.info("JSON is empty:" + taskDto.getMap().get("url"));
                                log.info(content);
                            }
                        } else {
                            log.info("JSON is empty:" + taskDto.getMap().get("url"));
                            log.info(content);
                        }
                    } else {
                        log.info("JSON is empty:" + taskDto.getMap().get("url"));
                        log.info(content);
                    }

                }catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }
}
